package com.example.iot.queue.rabbitMq;

import com.example.iot.queue.QueueConsumer;
import com.example.iot.queue.QueueFactory;
import com.example.iot.queue.QueueProducer;
import com.example.iot.queue.configuration.RuleEngineQueueConfig;
import com.example.iot.queue.message.ProtoQueueMsg;
import com.example.iot.queue.message.RuleEngineQueueMsg;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;

/**
 * @author ChenXiangpeng
 */
@Slf4j
@Component
@ConditionalOnExpression("'${queue.type:null}'=='rabbitmq'")
public class RabbitMqQueueFactory implements QueueFactory {

    private final RuleEngineQueueConfig ruleEngineQueueConfig;
    private final TbRabbitMqSettings rabbitMqSettings;

    private final TbRabbitMqAdmin ruleEngineAdmin;

    public RabbitMqQueueFactory(RuleEngineQueueConfig ruleEngineQueueConfig,
                                TbRabbitMqSettings taRabbitMqSettings,
                                TbRabbitMqQueueArguments queueArguments) {
        this.rabbitMqSettings = taRabbitMqSettings;

        this.ruleEngineQueueConfig = ruleEngineQueueConfig;

        this.ruleEngineAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getRuleEngineArgs());
    }

    @Override
    public QueueProducer<ProtoQueueMsg<RuleEngineQueueMsg>> createRuleEngineProducer() {
        return new TbRabbitMqProducerTemplate<>(ruleEngineAdmin, rabbitMqSettings, ruleEngineQueueConfig.getTopic());
    }

    @Override
    public QueueConsumer<ProtoQueueMsg<RuleEngineQueueMsg>> createRuleEngineConsumer() {
        return new TbRabbitMqConsumerTemplate<>(ruleEngineAdmin, rabbitMqSettings, ruleEngineQueueConfig.getTopic(),
                msg -> new ProtoQueueMsg<>(msg.getKey(), RuleEngineQueueMsg.parseFrom(msg.getData())));
    }

    @PreDestroy
    private void destroy() {
        if (ruleEngineAdmin != null) {
            ruleEngineAdmin.destroy();
        }
    }
}
